package WindowFun;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Iterator;
import java.util.List;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.stream.Collectors;

public class Press_WindowPro extends ProcessWindowFunction<ClassObject.Press, ClassObject.PressRes, Tuple, TimeWindow> {
    @Override
    public void process(Tuple tuple, Context context, Iterable<ClassObject.Press> elements, Collector<ClassObject.PressRes> out) throws Exception {
        String dateTime = ((Tuple1<String>)tuple).f0;
        //Java中的大小顶堆可以使用优先级队列来实现
        //https://blog.csdn.net/hefenglian/article/details/81807527
        //注意:
        // 小顶堆用来计算:最大的topN
        // 大顶堆用来计算:最小的topN
        Queue<ClassObject.Press> queue = new PriorityQueue<>(3,//初识容量
                //正常的排序,就是小的在前,大的在后,也就是c1>c2的时候返回1,也就是小顶堆
                (c1, c2) -> c1.getNum() >= c2.getNum() ? 1 : -1);

        //在这里我们要完成需求:
        // * -1.实时计算出11月11日00:00:00零点开始截止到当前时间的销售总额,其实就是把之前的初步聚合的price再累加!
        long sum=0L;
        Iterator<ClassObject.Press> iterator = elements.iterator();
        for (ClassObject.Press element : elements) {
            long num=element.num;//计算总销量
            sum+=num;
            if (queue.size() < 10) {//小顶堆size<3,说明数不够,直接放入
                queue.add(element);
            }else{//小顶堆size=3,说明,小顶堆满了,进来一个需要比较
                //"取出"顶上的(不是移除)
                ClassObject.Press top = queue.peek();
                if(element.num > top.num){
                    //queue.remove(top);//移除指定的元素
                    queue.poll();//移除顶上的元素
                    queue.add(element);
                }
            }
        }
        // * -3.每1秒钟更新一次统计结果,可以直接打印/sink,也可以收集完结果返回后再打印,
        //   但是我们这里一次性处理了需求1和2的两种结果,不好返回,所以直接输出!
        //对queue中的数据逆序
        //各个分类的销售额top3
        List<String> top3Result = queue.stream()
                .sorted((c1, c2) -> c1.getNum() > c2.getNum() ? -1 : 1)//逆序
                .map(c -> c.getPressname() + "," + c.getNum())
                .collect(Collectors.toList());
        System.out.println("时间 ： " + dateTime + "  总量 : " + sum + " top3:\n" + StringUtils.join(top3Result, ",\n"));
        System.out.println("===========");
        ClassObject.PressRes result3=new ClassObject.PressRes(top3Result);
        out.collect(result3);
    }
}
